/*
 * @(#)AgentTestsetQueue.java
 *
 *
 *
 * ver 1.0 Sep 11, 2007 ggranum
 */
package org.testfire.realm.control.agent;

import java.sql.Timestamp;
import java.util.*;
import javax.annotation.Resource;
import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.transaction.Status;
import javax.transaction.UserTransaction;
import org.apache.log4j.Logger;
import org.testfire.realm.control.agent.event.AgentAvailableEvent;
import org.testfire.realm.control.agent.event.AgentHasComeAvailableListener;
import org.testfire.realm.control.testfire.TestFireDBAO;
import org.testfire.realm.model.entities.testfire.*;

/**
 * @todo This should be converted to dispatch through a configurable Thread manager.
 */
public class AgentManager
{

    // Inject the lgsrealm entity manager, as defined in the persistence.xml file.
    @PersistenceContext(unitName = "testfire")
    EntityManager em;

    @Resource
    private UserTransaction utx;

    private static final PriorityQueue<AgentHasComeAvailableListener> agentAvailableListeners
        = new PriorityQueue<AgentHasComeAvailableListener>( 20, AgentHasComeAvailableListener.priorityComparator );

    static
    {
        agentAvailableListeners.offer( new AgentHasComeAvailableListener()
        {
            public void agentHasComeAvailable( AgentAvailableEvent event )
            {
                Logger log = Logger.getLogger( AgentManager.class );
                log.info( "Agent " + ((AgentEntity)event.getSource()).getId() + " has come available." );
                AgentManager am = TestFireDBAO.getValueExpresionValue( AgentManager.class );
                boolean b = am.scanAndRunIfAvailable();
                if ( b )
                {
                    log.info(
                        "The default agentHasComeAvailableEvent has caused at least one testset to be sent to an agent" );
                }
                else
                {
                    log.info( "The default agentHasComeAvailableEvent did not manage to trigger any testsets to be"
                        + " sent to any agents; this may be because there are none queued to run." );
                }
            }
        } );
    }


    Logger log = Logger.getLogger( this.getClass() );

    public AgentManager()
    {

    }

    private void enqueueTestset( TestRunnerQueueEntity queue, TestsetEntity testset )
    {
        TestsetResultEntity testsetResults = new TestsetResultEntity();
        testsetResults.setQueue( queue );
        testsetResults.setTestset( testset );
        testsetResults.setStatuscode( ResultStatus.WAITING_TO_START );
        try
        {
            boolean requiresTransaction = utx.getStatus() == Status.STATUS_NO_TRANSACTION;
            if(requiresTransaction)
                utx.begin();

            em.persist( testsetResults );
            if(requiresTransaction)
            utx.commit();
        }
        catch ( Exception ex )
        {
            log.error( "", ex);
        }
        scanAndRunIfAvailable();
    }

    public synchronized void enqueueTestset( TestRunnerQueueEntity queue )
    {
        for ( TestsetEntity set : queue.getTestsets() )
        {
            enqueueTestset( queue, set );
        }
        try
        {
            boolean requiresTransaction = utx.getStatus() == Status.STATUS_NO_TRANSACTION;
            if(requiresTransaction)
                utx.begin();
            em.merge( queue );
            if(requiresTransaction)
            utx.commit();
        }
        catch ( Exception ex )
        {
            log.error( "", ex);
        }
    }

    /**
     * First does a global check to see if there are any agents available AT ALL.
     * If so, loops through each waiting TestsetResultEntity with status WAITING_TO_START (if any) and
     * checks against the TestRunnerQueue for which agents are usable for the requested run.
     * Then checks if one of those AgentEntitys is available by first checking if the AgentEntity owns a TestsetResult
     * already, and then by pinging the agent to ensure that it is alive.
     * The TestsetResultEntity will then send out the TestsetEntity to the first AgentEntity that passes.
     *
     * @return True if a task was started.
     */
    public synchronized boolean scanAndRunIfAvailable()
    {

        // are there available agents?

        try
        {
            boolean requiresTransaction = utx.getStatus() == Status.STATUS_NO_TRANSACTION;
            if(requiresTransaction)
                utx.begin();
            List<AgentEntity> availableAgentsList = new ArrayList<AgentEntity>( TestFireDBAO.getAvailableAgents(em) );
            Map<Integer, AgentEntity> availableAgents = new HashMap<Integer, AgentEntity>();
            for ( AgentEntity a : availableAgentsList )
            {
                availableAgents.put( a.getId(), a );
            }
            List<AgentEntity> locallyAvailableAgents;

            if ( availableAgents.size() == 0 )
            {
                log.info( "No agents available to run testsets on." );
                return false;
            }
            // we have agents to use; do we have testresults with status == waiting?
            List<TestsetResultEntity> testsetResults = new ArrayList<TestsetResultEntity>(
                TestFireDBAO.getTestsetResultByStatus( ResultStatus.WAITING_TO_START, em ) );
            if ( testsetResults.size() == 0 )
            {
                log.info( "No testsets queued to run." );
                return false;
            }

            // we have agents, we have testsets.
            for ( int i = 0; i < testsetResults.size(); i++ )
            {
                // Get the queue, so we can get the ALLOWED agents.
                TestsetResultEntity result = testsetResults.get( i );
                TestRunnerQueueEntity queue = result.getQueue();
                // Available agents:
                if ( !queue.getUseAnyAvailableAgent() )
                {
                    locallyAvailableAgents = new ArrayList<AgentEntity>( queue.getAgents() );
                }
                else
                {
                    locallyAvailableAgents = availableAgentsList;
                }
                for ( int k = 0; k < locallyAvailableAgents.size(); k++ )
                {
                    AgentEntity a = locallyAvailableAgents.get( k );
                    if ( !availableAgents.containsKey( a.getId() ) )
                    {
                        locallyAvailableAgents.remove( k );
                    }
                }
                for ( int j = 0; j < locallyAvailableAgents.size(); j++ )
                {
                    AgentEntity agent = locallyAvailableAgents.get( j );
                    boolean available = pingAgent( agent );
                    if ( !available )
                    {
                        log.info( "Agent " + agent.getId() + " is not available: ping failed" );
                        availableAgents.remove( agent.getId() );
                        continue;
                    }
                    log.info( "Sending testset " + result.getTestset().getName() + " to agent: " + agent.getId() );
                    boolean accepted = sendToAgent( agent, result.getTestset(), result );
                    log.info( "Testset " + result.getTestset().getName() + " was sent to agent: "
                        + agent.getId() + ". Accepted: " + accepted );
                    if ( accepted )
                    {
                        availableAgents.remove( agent.getId() );
                        TestsetResultEntity tsr = testsetResults.remove( j );
                        agent.setLastPingTime( System.currentTimeMillis() );
                        agent.setHastestset( tsr.getTestset() );
                        tsr.setStarted( new Timestamp( System.currentTimeMillis() ) );
                        tsr.setStatuscode( ResultStatus.RUNNING );
                        em.merge( agent );
                        em.merge( tsr );
                        em.flush();
                        log.info( "Testset: " + tsr.getTestset().getName() + " updated after submission to agent." );
                    }
                }

            }
           if(requiresTransaction)
            utx.commit();
        }
        catch ( Exception e )
        {
            log.error( "Scan and run failed:", e );
            return false;
        }

        return true;
    }

    public synchronized boolean sendToAgent( AgentEntity agent, TestsetEntity testset, TestsetResultEntity result )
    {
        return ServerToAgentComm.sendTestsetToAgent( agent, testset, result );
    }

    public boolean pingAgent( AgentEntity agent )
    {
        int delta = ServerToAgentComm.pingAgent( agent );
        agent.setLastping( delta );
        return delta >= 0;
    }

    /**
     * Listen for agents becoming available; if an agent comes available and the listener chooses
     * to claim it, the listener MUST consume the event.
     * <p/>
     * Additionally, all listeners that consume an event must do so in the invoking thread, blocking the completion of
     * AgentHasComeAvailableListener.agentHasComeAvailable(event) until the event has been marked as consumed.
     *
     * @param listener The listener.
     */
    public synchronized void addAgentHasComeAvailableListener( AgentHasComeAvailableListener listener )
    {
        listener.setPriority( agentAvailableListeners.size() );// start at the back of the line.
        agentAvailableListeners.offer( listener );
    }

    public synchronized void fireAgentHasComeAvailableEvent( AgentAvailableEvent event )
    {
        AgentHasComeAvailableListener[] ary =
            agentAvailableListeners.toArray( new AgentHasComeAvailableListener[agentAvailableListeners.size()] );
        int i = 0;
        for ( ; i < ary.length; i++ )
        {
            if ( event.isConsumed() )
            {
                ary[i].increasePriority();
            }
            else
            {
                ary[i].agentHasComeAvailable( event );
                if ( event.isConsumed() )
                {
                    // set the priority to the size of the array: move to the back of the line.
                    ary[i].setPriority( ary.length );
                }
            }
        }
    }


}

// eof: AgentTestsetQueue.java